package com.wuuxiang.polestar.core.zookeeper.surpass;

import java.io.IOException;
import java.util.List;
import java.util.Map;
import java.util.Set;

import org.apache.commons.io.FileUtils;
import org.apache.commons.lang3.tuple.ImmutablePair;
import org.apache.commons.lang3.tuple.Pair;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.api.CuratorListener;
import org.apache.curator.framework.api.GetChildrenBuilder;
import org.apache.curator.framework.api.GetDataBuilder;
import org.apache.curator.utils.ZKPaths;
import org.apache.zookeeper.data.Stat;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.google.common.base.Charsets;
import com.google.common.base.Throwables;
import com.google.common.collect.Maps;
import com.wuuxiang.polestar.comm.constant.Constant;
import com.wuuxiang.polestar.core.ConfigGroup;
import com.wuuxiang.polestar.core.GeneralConfigGroup;

/**
 * 项目名称：polestar        <br>
 * 类描述：                         <br>
 * 创建人：夏鸿鹏                    <br>
 * 创建时间：2017-12-5 15:39        <br>
 * 修改备注：                        <br>
 */

public class ZookeeperSurPassConfigGroup extends GeneralConfigGroup {
    private static final long serialVersionUID = 1L;

    static final Logger LOGGER = LoggerFactory.getLogger(ZookeeperSurPassConfigGroup.class);

    // 配置对象
    private ZookeeperSurPassConfigProfile configProfile;
    // 节点名称
    private String node;
    // ZK客户端
    private CuratorFramework client;
    // 缓存对象
    private ConfigLocalSurPassCache configLocalCache;
    // 节点数据变更监听
    private CuratorListener listener = new ConfigNodeSurPassEventListener(this);

    public ZookeeperSurPassConfigGroup(ZookeeperSurPassConfigProfile configProfile, String node) {
        this(null, configProfile, node);
    }

    public ZookeeperSurPassConfigGroup(ConfigGroup internalConfigGroup,
                                       ZookeeperSurPassConfigProfile configProfile, String node) {
        super(internalConfigGroup);
        this.configProfile = configProfile;
        this.node = node;

        if (configProfile.isOpenLocalCache()) {
            configLocalCache = new ConfigLocalSurPassCache(
                    FileUtils.getUserDirectoryPath() + "/.polestar",
                    configProfile.getTeamName());
        }
        initConfigs();
    }

    /**
     * 初始化节点
     */
    private void initConfigs() {
        client = CuratorFrameworkFactory.newClient(configProfile.getConnectStr(), configProfile.getRetryPolicy());
        client.start();
        client.getCuratorListenable().addListener(listener);

        LOGGER.info(
                "Loading properties for node: {}, with loading mode: {} and keys specified: {}",
                node, configProfile.getKeyLoadingMode(),
                configProfile.getKeysSpecified());
        loadNode();
        // Update local cache
        if (configLocalCache != null) {
            configLocalCache.saveLocalCache(this, node);
        }
    }

    /**
     * 加载节点并监听节点变化
     */
    void loadNode() {
        // 获取配置文件zk路径
        String nodePath = ZKPaths.makePath(Constant.ROOT_NODE, Constant.CONFIG_NODE, configProfile.getTeamName() , node);
        // 获取默认配置文件目录
        String defNodePath = ZKPaths.makePath(Constant.ROOT_NODE, Constant.CONFIG_NODE, Constant.DEF_NODE , node);
        GetChildrenBuilder childrenBuilder = client.getChildren();

        try {
            // 如果启用了默认节点加载则先加载默认节点
            List<String> defChildren = null;
            if (configProfile.isLoadDef()) {
            	Stat stat = client.checkExists().forPath(defNodePath);
            	if (stat != null) {
            		defChildren = childrenBuilder.watched().forPath(defNodePath);
            	}
            }

            // 如果在项目路径下存在同名配置文件则覆盖default下的配置
            Stat stat = client.checkExists().forPath(nodePath);
            if (stat != null) {
            	// 取出配置文件下所有配置项节点
            	List<String> children = childrenBuilder.watched().forPath(nodePath);
            	// 如果项目组节点下和默认项目组先配置文件重名,则覆盖默认组配置
            	if (children != null) {
            		defChildren = children;
            		defNodePath = nodePath;
            	}
            }

            if (defChildren != null) {
                final Map<String, String> configs = Maps.newHashMap();
                // 循环写入配置项value
                for (String child : defChildren) {
                    final Pair<String, String> keyValue = loadKey(ZKPaths.makePath(defNodePath, child));
                    if (keyValue != null) {
                        configs.put(keyValue.getKey(), keyValue.getValue());
                    }
                }
                cleanAndPutAll(configs);
            }
        } catch (Exception e) {
            LOGGER.error("加载节点并监听异常",e);
            throw Throwables.propagate(e);
        }
    }

    void reloadKey(final String nodePath) {
        try {
            final Pair<String, String> keyValue = loadKey(nodePath);
            if (keyValue != null) {
                super.put(keyValue.getKey(), keyValue.getValue());
            } else {
                LOGGER.warn("keyValue为空");
            }
        } catch (Exception e) {
            LOGGER.error("加载变换的key异常:"+e.getMessage(),e);
            throw Throwables.propagate(e);
        }
    }

    private Pair<String, String> loadKey(final String nodePath)
            throws Exception {
        final String nodeName = ZKPaths.getNodeFromPath(nodePath);
        final Set<String> keysSpecified = configProfile.getKeysSpecified();
        switch (configProfile.getKeyLoadingMode()) {
            case INCLUDE:
                if (keysSpecified == null || !keysSpecified.contains(nodeName)) {
                    return null;
                }
                break;
            case EXCLUDE:
                if (keysSpecified.contains(nodeName)) {
                    return null;
                }
                break;
            case ALL:
                break;
            default:
                break;
        }

        final GetDataBuilder data = client.getData();
        final String value = new String(data.watched().forPath(nodePath),
                Charsets.UTF_8);
        return new ImmutablePair<String, String>(nodeName, value);
    }

    public String getNode() {
        return node;
    }

    public ConfigLocalSurPassCache getConfigLocalCache() {
        return configLocalCache;
    }

    /**
     * 导出属性列表
     *
     * @return
     */
    public Map<String, String> exportProperties() {
        return Maps.newHashMap(this);
    }

    @Override
    public void close() throws IOException {
        if (client != null) {
            client.getCuratorListenable().removeListener(listener);
            client.close();
        }
    }

}
